package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.DimAsyncFunction;
import com.atguigu.bean.CourseCommentBean;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Dws132_CourseCommentWindow {
    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取KafkaDWD层评论主题数据
        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("dwd_interaction_comment_info", "course_comment"), WatermarkStrategy.noWatermarks(), "kafka-source");

        //3.过滤Null值并转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    out.collect(JSON.parseObject(value));
                }
            }
        });

        //4.提取时间戳生成Watermark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWMDS = jsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                System.out.println("watermark>>>>>>>>>"+element.getLong("ts"));

                return element.getLong("ts");
            }
        }));
        jsonObjWithWMDS.print("jsonObjWithWMDS-->");


        //5.按照用户ID分组,并直接转换为JavaBean对象
        SingleOutputStreamOperator<CourseCommentBean> courseCommentDS = jsonObjWithWMDS
                .keyBy(json -> json.getString("user_id"))
                .flatMap(new RichFlatMapFunction<JSONObject, CourseCommentBean>() {

                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("comment-state", String.class));
                    }

                    @Override
                    public void flatMap(JSONObject value, Collector<CourseCommentBean> out) throws Exception {

                        //取出状态数据
                        String state = valueState.value();
                        String curDt = value.getString("ts").split(" ")[0];

                        long cuc = 0L;
                        long goodCuc = 0L;

                        if (state == null || !state.equals(curDt)) {
                            cuc = 1L;
                            valueState.update(curDt);
                        }
                        if (value.getLong("review_stars") == 5) {
                            goodCuc = 1L;
                            valueState.update(curDt);
                        }

                        if (cuc == 1L) {
                            out.collect(new CourseCommentBean(
                                    "",
                                    "",
                                    value.getString("course_id"),
                                    "",
                                    value.getString("user_id"),
                                    value.getLong("review_stars"),
                                    cuc,
                                    goodCuc,
                                    value.getLong("create_time")));
                        }
                    }
                });

        courseCommentDS.print("courseCommentDS-->");

        //6.分组、开窗、聚合
        SingleOutputStreamOperator<CourseCommentBean> reduceDS = courseCommentDS.keyBy(CourseCommentBean::getCourseId)
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<CourseCommentBean>() {
                    @Override
                    public CourseCommentBean reduce(CourseCommentBean value1, CourseCommentBean value2) throws Exception {
                        value1.setUserReview(value1.getUserReview() + value2.getUserReview());
                        value1.setCommentUserCount(value1.getCommentUserCount() + value2.getCommentUserCount());
                        value1.setGoodUserReviewCount(value1.getGoodUserReviewCount() + value2.getGoodUserReviewCount());
                        System.out.println("1");
                        return value1;
                    }
                }, new WindowFunction<CourseCommentBean, CourseCommentBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<CourseCommentBean> input, Collector<CourseCommentBean> out) throws Exception {

                        CourseCommentBean next = input.iterator().next();

                        next.setTs(System.currentTimeMillis());
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        System.out.println("2");
                        out.collect(next);
                    }
                });
        reduceDS.print("reduceDS-->");

        //7.关联维表,补充课程名称
        SingleOutputStreamOperator<CourseCommentBean> resultDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<CourseCommentBean>("DIM_COURSE_INFO") {
            @Override
            public String getkey(CourseCommentBean input) throws Exception {
                return input.getCourseId();
            }

            @Override
            public void join(CourseCommentBean input, JSONObject dimInfo) throws Exception {
                input.setCourseName(dimInfo.getString("COURSE_NAME"));
            }
        }, 60, TimeUnit.SECONDS);

        //8.将数据写出
        resultDS.print("resultDS>>>>>");
        resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_course_comment_window values(?,?,?,?,?,?,?,?)"));

        //9.启动任务
        env.execute("Dws132_CourseCommentWindow");
    }
}
